package demo.kafka.simple;


import org.apache.kafka.clients.producer.RecordMetadata;

import org.apache.kafka.clients.producer.Callback;

/**
 * kafka消息发送成功的回调类
 *
 * @author 千里明月
 * @date 2021/8/4 23:07
 */
public class ProducerCallback implements Callback {

    private long startTime;
    private int key;
    private String message;

    public ProducerCallback(long startTime, int key, String message) {
        this.startTime = startTime;
        this.key = key;
        this.message = message;
    }

    /**
     * 回调方法
     *
     * @author 千里明月
     * @date 2021/8/9
     **/
    @Override
    public void onCompletion(RecordMetadata metadata, Exception e) {
        long duration = System.currentTimeMillis() - startTime;
        if (metadata != null) {
            System.out.println("key=" + key + ",value=" + message + ",partition=" + metadata.partition() + ",offset=" + metadata.offset() + ",duration=" + duration);
        } else {
            e.printStackTrace();
        }
    }

}
